package messagesvc.websocket.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import cn.ucloud.microsf.messagesvc.util.RedisUtil;
import cn.ucloud.microsf.messagesvc.util.RedisUtil.ListOps;
import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

@Slf4j
public class WsSession {

  public static final String SESSION_KEY = "token";
  private static final String RECEIVER_CACHE = "SF:MICRO:MESSAGE:SVC:WS:RECEIVER:";

  private static final ConcurrentHashMap<String, WebSocketSession> SESSION_POLL = new ConcurrentHashMap<>();

  public static void add(String key, WebSocketSession session) {
    if (!SESSION_POLL.containsKey(key)) {
      log.info("连接成功-新增:【{}】 ", key);
      SESSION_POLL.put(key, session);
    } else {
      WebSocketSession existSession = get(key);
      if (!StrUtil.equals(session.getId(), existSession.getId())) {
        log.info("连接已存在-刷新:【{}】 ", key);
        SESSION_POLL.put(key, session);
      }
    }

  }

  public static void add(WebSocketSession session) {
    add(getToken(session), session);
  }

  public static void remove(String key) {
    WebSocketSession session = SESSION_POLL.remove(key);
    if (session != null) {
      try {
        session.close();
        log.info("断开链接:【{}】 ", key);
      } catch (IOException e) {
        log.error("", e);
        log.info("断开链接失败:【{}】 ", key);
      }
    }
  }

  public static void remove(WebSocketSession session) {
    remove(getToken(session));
  }

  public static String getToken(WebSocketSession session) {
    Object token = session.getAttributes().get(SESSION_KEY);
    if (token != null) {
      return token.toString();
    }
    return null;
  }

  public static WebSocketSession get(String key) {
    return SESSION_POLL.get(key);
  }

  public static void getAndSend(String key, String text) {
    try {
      WebSocketSession session = get(key);
      if (Objects.nonNull(session)) {
        session.sendMessage(new TextMessage(text));
      } else {
        // 放入缓存 等接收方连接时发送
        RedisUtil.ListOps.lRightPush(RECEIVER_CACHE + key, text);
      }
    } catch (Exception e) {
      log.error("", e);
    }
  }

  public static void readCacheAndSend(String key) {
    try {
      WebSocketSession session = get(key);
      if (Objects.nonNull(session)) {
        String cacheKey = RECEIVER_CACHE + key;
        if (RedisUtil.KeyOps.hasKey(cacheKey)) {
          List<String> contents = ListOps.lWholeList(cacheKey);
          if (CollectionUtil.isNotEmpty(contents)) {
            for (String content : contents) {
              session.sendMessage(new TextMessage(content));
            }
          }
          RedisUtil.KeyOps.delete(cacheKey);
        }
      }
    } catch (Exception e) {
      log.error("", e);
    }
  }

}
